package com.open.source;

import com.open.source.kafka.ApplicationContextHolder;
import java.util.function.Consumer;
import lombok.Getter;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.TopicExchange;

/**
 * @author ZonLen since on 2022/8/21 上午1:22
 */
public enum MqManagerEnum implements MqConstant {

  INVOICE_TOPIC__CALLBACK_WEBSERVICE(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_WEBSERVICE, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_WEBSERVICE_QUEUE,
          "#.webservice.#")),
  /**
   * 红字信息申请表购方申请回调推送通道
   */
  INVOICE_TOPIC__CALLBACK_RED_APPLY(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_RED_APPLY, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_RED_APPLY_QUEUE,
          "#.redapply.#")),
  /**
   * 发票信息推送通道
   */
  INVOICE_TOPIC__CALLBACK_GATHER_PUSH(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_GATHER_PUSH, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_GATHER_PUSH_QUEUE,
          "#.gatherpush.#")),

  INVOICE_TOPIC__CALLBACK_RESTFUL(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_RESTFUL, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_RESTFUL_QUEUE,
          "#.restful.#")),

  INVOICE_TOPIC__CALLBACK_EMAIL(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_EMAIL, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_EMAIL_QUEUE, "#.email.#")),

  INVOICE_TOPIC__CALLBACK_SMS(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_SMS, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_SMS_QUEUE, "#.sms.#")),
  INVOICE_TOPIC__CALLBACK_WEIXIN(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_WEIXIN, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_WEI_XIN_QUEUE,
          "#.weixin.#")),
  INVOICE_TOPIC__CALLBACK_CORP_REV(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_CORP_REV, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_CORP_REV_QUEUE,
          "#.corpRev.#")),
  INVOICE_TOPIC__CALLBACK_COUNT_TAXMECH(
      new KafkaListenerConstant(MqConstant.INVOICE_TOPIC__CALLBACK_COUNT_TAXMECH, null),
      new RabbitRouting(RABBIT_EINVOICE_TOPIC_EXCHANGE, RABBIT_CALLBACK_COUNT_TAX_MECH_QUEUE,
          "#.count-invoice.#")),

  /**
   * 发票回传结果处理延时队列
   */
  INVOICE_DIRECT__INVOICE_CLIENT_DELAY(
      new KafkaListenerConstant(MqConstant.INVOICE_DIRECT__INVOICE_CLIENT_DELAY, null),
      new RabbitRouting(RABBIT_EINVOICE_DIRECT_EXCHANGE, RABBIT_INVOICE_CLIENT_DELAY_QUEUE,
          RABBIT_EINVOICE_DIRECT_EXCHANGE)
  ),
  INVOICE_DIRECT__INVOICE_CLIENT_ERROR(
      new KafkaListenerConstant(MqConstant.INVOICE_DIRECT__INVOICE_CLIENT_ERROR, null),
      new RabbitRouting(RABBIT_EINVOICE_DIRECT_EXCHANGE, RABBIT_INVOICE_CLIENT_ERROR_QUEUE,
          RABBIT_INVOICE_CLIENT_ERROR_QUEUE)
  ),
  INVOICE_DIRECT__INVOICE_CLIENT(
      new KafkaListenerConstant(MqConstant.INVOICE_DIRECT__INVOICE_CLIENT, null),
      new RabbitRouting(RABBIT_EINVOICE_DIRECT_EXCHANGE, RABBIT_INVOICE_CLIENT_QUEUE,
          RABBIT_INVOICE_CLIENT_QUEUE)
  ),

  ERECEIPT_TOPIC__CALLBACK_RESTFUL_DELAY(
      new KafkaListenerConstant(MqConstant.ERECEIPT_TOPIC__CALLBACK_RESTFUL_DELAY, null),
      new RabbitRouting(RABBIT_ERECEIPT_TOPIC_EXCHANGE,
          RABBIT_ERECEIPT_CALLBACK_RESTFUL_DELAY_QUEUE,
          "#.restful.#")
  ),
  ERECEIPT_TOPIC__CALLBACK_RESTFUL_ERROR(
      new KafkaListenerConstant(MqConstant.ERECEIPT_TOPIC__CALLBACK_RESTFUL_ERROR, null),
      new RabbitRouting(RABBIT_ERECEIPT_TOPIC_EXCHANGE,
          RABBIT_ERECEIPT_CALLBACK_RESTFUL_ERROR_QUEUE,
          "#.restful.#")
  ),
  ERECEIPT_TOPIC__CALLBACK_RESTFUL(
      new KafkaListenerConstant(MqConstant.ERECEIPT_TOPIC__CALLBACK_RESTFUL, null),
      new RabbitRouting(RABBIT_ERECEIPT_TOPIC_EXCHANGE, RABBIT_ERECEIPT_CALLBACK_RESTFUL_QUEUE,
          "#.restful.#")
  ),


  ERECEIPT_DIRECT__INVOICING_APPLY_DELAY(
      new KafkaListenerConstant(MqConstant.ERECEIPT_DIRECT__INVOICING_APPLY_DELAY, null),
      new RabbitRouting(RABBIT_ERECEIPT_DIRECT_EXCHANGE, RABBIT_ERECEIPT_APPLY_DELAY_QUEUE,
          RABBIT_ERECEIPT_APPLY_DELAY_QUEUE)
  ),
  ERECEIPT_DIRECT__INVOICING_APPLY_ERROR(
      new KafkaListenerConstant(MqConstant.ERECEIPT_DIRECT__INVOICING_APPLY_ERROR, null),
      new RabbitRouting(RABBIT_ERECEIPT_DIRECT_EXCHANGE, RABBIT_ERECEIPT_APPLY_ERROR_QUEUE,
          RABBIT_ERECEIPT_APPLY_ERROR_QUEUE)
  ),
  ERECEIPT_DIRECT__INVOICING_APPLY(
      new KafkaListenerConstant(MqConstant.ERECEIPT_DIRECT__INVOICING_APPLY, null),
      new RabbitRouting(RABBIT_ERECEIPT_DIRECT_EXCHANGE, RABBIT_ERECEIPT_APPLY_QUEUE,
          RABBIT_ERECEIPT_APPLY_QUEUE)
  ),

  UAP_DIRECT__INVOICE_BASE(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__INVOICE_BASE, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_INVOICE_BASE_QUEUE,
          "invoice_base_queue_key")
  ),

  UAP_DIRECT__INVOICE_OFD_DELAY(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__INVOICE_OFD_DELAY, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_INVOICE_OFD_DELAY_QUEUE,
          RABBIT_UAP_INVOICE_OFD_QUEUE)
  ),

  UAP_DIRECT__INVOICE_OFD_ERROR(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__INVOICE_OFD_ERROR, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_INVOICE_OFD_ERROR_QUEUE,
          RABBIT_UAP_INVOICE_OFD_ERROR_QUEUE)
  ),

  UAP_DIRECT__INVOICE_OFD(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__INVOICE_OFD, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_INVOICE_OFD_QUEUE,
          RABBIT_UAP_INVOICE_OFD_QUEUE)
  ),

  UAP_DIRECT__BILLING_LOG(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__BILLING_LOG, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_BILLING_LOG_QUEUE,
          RABBIT_UAP_BILLING_LOG_QUEUE)
  ),
  UAP_DIRECT__BILL_CENT_SYNC(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__BILLING_LOG, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_BILL_CENT_SYNC_QUEUE,
          "bill-center-sync-key")
  ),
  UAP_DIRECT__MESSAGE_PDF(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__MESSAGE_PDF, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_MESSAGE_PDF_QUEUE,
          RABBIT_UAP_MESSAGE_PDF_QUEUE)
  ),
  UAP_DIRECT__CUSTOMER_SYNC(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__CUSTOMER_SYNC, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_CUSTOMER_SYNC_QUEUE,
          "customer_sync_queue_key")
  ),
  UAP_DIRECT__ADMISSION_CONSUME(
      new KafkaListenerConstant(MqConstant.UAP_DIRECT__ADMISSION_CONSUME, null),
      new RabbitRouting(RABBIT_UAP_DIRECT_EXCHANGE, RABBIT_UAP_ADMISSION_CONSUME_QUEUE,
          RABBIT_UAP_ADMISSION_CONSUME_QUEUE)
  ),


  TAX_TOPIC_DELAY_DEMO(new KafkaListenerConstant(MqConstant.TAX_TOPIC_DELAY_DEMO, null),
      new RabbitRouting("", "", "")),

  TAX_TOPIC_DEMO(new KafkaListenerConstant(MqConstant.TAX_TOPIC_DEMO, null),
      new RabbitRouting("", "", ""));


  private final KafkaListenerConstant kafkaListenerConstant;

  private final RabbitRouting rabbitRouting;

  MqManagerEnum(KafkaListenerConstant kafkaListenerConstant, RabbitRouting rabbitRouting) {
    this.kafkaListenerConstant = kafkaListenerConstant;
    this.rabbitRouting = rabbitRouting;
  }

  public String getTopic() {
    return ApplicationContextHolder.getEnvironment()
        .resolvePlaceholders(kafkaListenerConstant.kafkaTopic);
  }

  public RabbitRouting getRabbitRouting() {
    return rabbitRouting;
  }


  public static class KafkaListenerConstant {

    private final String kafkaTopic;

    private final String groupId;

    public KafkaListenerConstant(String kafkaTopic, String groupId) {
      this.kafkaTopic = kafkaTopic;
      this.groupId = groupId;
    }
  }


  public static class RabbitRouting {

    @Getter
    private final String exchange;
    @Getter
    private final String queue;
    @Getter
    private final String routingKey;

    private boolean isDelay;

    public RabbitRouting(String exchange, String queue, String routingKey) {
      this(exchange, queue, routingKey, false);
    }

    public RabbitRouting(String exchange, String queue, String routingKey, boolean isDelay) {
      this.exchange = exchange;
      this.queue = queue;
      this.routingKey = routingKey;
      this.isDelay = isDelay;
    }

    public Queue queue() {
      return new Queue(this.queue, true, false, false);
    }

    public Queue queue(Consumer<QueueBuilder> consumer) {
      final QueueBuilder queueBuilder = QueueBuilder.durable(queue);
      if (null != consumer) {
        consumer.accept(queueBuilder);
      }
      return queueBuilder.build();
    }

    public Binding bindingDirectExchange(Queue queue, DirectExchange exchange) {
      return BindingBuilder.bind(queue).to(exchange).with(routingKey);
    }

    public Binding bindingTopicExchange(Queue queue, TopicExchange exchange) {
      return BindingBuilder.bind(queue).to(exchange).with(routingKey);
    }

  }
}
